package com.workerthread;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author 罗俊华
 * @date 2021/8/24 - 2:30 下午
 */
@Slf4j
public class BossSelector {

    public static void main(String[] args) throws IOException {

        Selector bossSelector = Selector.open();

        Thread.currentThread().setName("boss 线程");

//        工作者线程池
        /*ExecutorService workerThreadPool = Executors.newCachedThreadPool(new ThreadFactory() {
            AtomicInteger count = new AtomicInteger();

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("工作者线程-" + count.incrementAndGet());
                return thread;
            }
        });*/

        WorkerTask[] workerTasks = new WorkerTask[3];

        for (int i = 0; i < workerTasks.length; i++) {
            workerTasks[i] = new WorkerTask("工作者线程" + i);

        }


        try (ServerSocketChannel serverSocketChannel = ServerSocketChannel.open()) {

            serverSocketChannel.bind(new InetSocketAddress(5689))
                    .configureBlocking(false);

            serverSocketChannel.register(bossSelector, SelectionKey.OP_ACCEPT);

            while (true) {

                log.debug("线程【{}】在等待 accept 事件", Thread.currentThread().getName());

                bossSelector.select();

                log.debug("监听到 accept 事件触发");

                Iterator<SelectionKey> selectionKeyIterator = bossSelector.selectedKeys().iterator();

                int index = 0;

                while (selectionKeyIterator.hasNext()) {

                    SelectionKey selectionKey = selectionKeyIterator.next();

                    selectionKeyIterator.remove();

                    if (selectionKey.isAcceptable()) {

                        ServerSocketChannel sscChannel = (ServerSocketChannel) selectionKey.channel();

                        SocketChannel socketChannel = sscChannel.accept();
                        index = index++ % workerTasks.length;
                        WorkerTask workerTask = workerTasks[index];

                        log.debug("准备将新 accept 的 channel 注册到 {}", workerTask.getWorkerName());

                        /**
                         * 如果 worker 线程中 this.selector.select();先启动
                         * 那么由于 selector 中没有事件，所以 select() 方法会被阻塞
                         * 此时如果 boos 线程 调用 socketChannel.register(this.selector, SelectionKey.OP_READ, ByteBuffer.allocate(8));
                         * 那么由于 select() 方法被阻塞，会影响到 register() 方法也会被阻塞，
                         * 最终导致：worker线程在等待事件的触发，此时占用着"锁"，而boss线程要注册 channel 来给 worker 线程监听事件，但也因为阻塞而无法注册
                         *
                         * 如果不做处理，无论如何一定会出现上述情况
                         * 当工作者线程处理完第一个读事件之后，工作者线程一定阻塞在 selector.select() 方法上
                         * 这时，新来一个 accept 事件，boos 线程试图往 工作者的 selector 调用 register() 注册一个 channel 时，就被阻塞了
                         */
                        workerTask.register(socketChannel);

                        log.debug("注册 channel 到 {} 成功", workerTask.getWorkerName());
                    }


                }


            }

        }

    }
}
